package ar.edu.unlu.sistemas.ws.business;

import java.io.IOException;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.net.InetSocketAddress;
import java.net.Socket;
import java.net.UnknownHostException;
import java.util.Collections;
import java.util.Comparator;
import java.util.LinkedList;
import java.util.List;
import java.util.Queue;

import org.apache.log4j.Logger;

import ar.edu.unlu.sistemas.ws.business.CORI.ResourceRank;

public class QueryPropagator {

	private static Logger logger = Logger.getLogger("net");
	
	private String query;
	private  List<ResourceRank> resourcesRank;
	private int finished = 0;
	private ResultMergeManager merge = new RoundRobinMerge();
	
	public QueryPropagator(String query, List<ResourceRank>resourcesRank) {
		this.query = query;
		this.resourcesRank = resourcesRank;
	}
	
	/**
	 * Propara la query a todos los peers
	 * @return resultados de la query distribuida
	 * @throws InterruptedException
	 */
	public List<QueryResultVO> propagateQuery() throws InterruptedException   {
		int launched = 0;
		for (ResourceRank peerRank : this.resourcesRank) {
			try {
				this.sendQuery(peerRank);
				launched++;
			} catch (Exception e) {
				logger.error("Ocurrio un error propagando la query \""+query+"\"", e);
			}
		}
		while (launched > this.finished) {
			Thread.sleep(5);
		}
		
		return merge.getResults();
		
	}
	
	/**
	 * Metodo para agregar resultados manejando concurrencia
	 * @param subResult
	 */
	private synchronized void addResult(String peerId, List<QueryResultVO> subResult, Double peerRank) {
		merge.addResult(peerId,subResult,peerRank);
	}
	
	/**
	 * Inicia la conexion con cada thread
	 * @param peerId
	 */
	private  void sendQuery(ResourceRank peerRank)  {
		QueryThread  thread = new QueryThread(peerRank);
		thread.start();
	}
	
	/**
	 * Inner class que es el thread que se conecta a cada peer
	 * @author pepo
	 *
	 */
	private class QueryThread extends Thread {
		
		private ResourceRank peerRank;
		
		public QueryThread(ResourceRank peerRank) {
			this.peerRank = peerRank;
		}
		
		public void run() {
			PeerWS peer = PeerManager.getInstance().getPeer(peerRank.peerId);
			Socket s = new Socket();
			try {
				logger.info("Inicio conexion para propagacion de query\""+query+"\" con "+peer.toString());
				InetSocketAddress address = new InetSocketAddress(peer.getAddress(),  peer.getPort());
				s.connect(address);
				s.setTcpNoDelay(true);
				s.setPerformancePreferences(2, 1, 0);
				logger.info("Se establecio conexion para propagacion de query\""+query+"\" con "+peer.toString());
				ObjectOutputStream out = new ObjectOutputStream(s.getOutputStream());
				logger.info("Tengo ouput de conexion para propagacion de query\""+query+"\" con "+peer.toString());
				Message msg = new Message();
				msg.setType("QUERY");
				msg.setQuery(query);
				logger.info("Preparo query\""+query+"\" a enviar a "+peer.toString());
				out.writeObject(msg);
				logger.info("Envio query\""+query+"\" a "+peer.toString());
				try {
					ObjectInputStream in = new ObjectInputStream(s.getInputStream());
					logger.info("Tengo input de conexion para propagacion de query\""+query+"\" con "+peer.toString());
					Message response = (Message) in.readObject();
					addResult(this.peerRank.peerId,response.getResult(),this.peerRank.rank);
				} catch (ClassNotFoundException cne) {
					logger.error("Ocurrio esperando la respuesta del movil para la query \""+query+"\"", cne);
				}
				logger.info("Cierro conexion de propagacion de query\""+query+"\" con "+peer.toString());
				s.close();
			} catch (UnknownHostException uhe) {
				logger.error("Ocurrio un error estableciendo conexión con "+peer.toString()+" para la query \""+query+"\"", uhe);
			} catch (IOException ioe) {
				logger.error("Ocurrio un erro I/O con "+peer.toString()+" para la query \""+query+"\"", ioe);
			}
			finished++;
		}
	};
	
	private static class QueryResultWrapper {
		QueryResultWrapper(String peerId, Double rank, List<QueryResultVO> results) {
			this.peerId = peerId;
			this.rank = rank;
			this.results = (Queue<QueryResultVO>)results;
		}
		String peerId;
		Double rank;
		Queue<QueryResultVO> results;
	}
	
	private static interface ResultMergeManager {
		void addResult(String peerId, List<QueryResultVO> subResult, Double peerRank);
		List<QueryResultVO> getResults();
	}
	private static class RoundRobinMerge implements ResultMergeManager {
		private List<QueryResultWrapper> list = new LinkedList<QueryResultWrapper>();
		public void addResult(String peerId, List<QueryResultVO> subResult, Double peerRank) {
			list.add(new QueryResultWrapper(peerId,peerRank,subResult));
		}
		public List<QueryResultVO> getResults() {
			Collections.sort(list, new Comparator<QueryResultWrapper>() {
				public int compare(QueryResultWrapper o1, QueryResultWrapper o2) {
					return o2.rank.compareTo(o1.rank);
				}
			});
			Queue<QueryResultWrapper> rr = (Queue<QueryResultWrapper>) list;
			List<QueryResultVO> results = new LinkedList<QueryResultVO>();
			QueryResultWrapper qrw;
			while (!rr.isEmpty()) {
				qrw = rr.poll();
				results.add(qrw.results.poll());
				if(!qrw.results.isEmpty()) rr.add(qrw);
			}	
			return results;
		}
	}
	private static class SortMerge implements ResultMergeManager {
		private List<QueryResultVO> list = new LinkedList<QueryResultVO>();
		public void addResult(String peerId, List<QueryResultVO> subResult, Double peerRank) {
			list.addAll(subResult);
		}
		public List<QueryResultVO> getResults() {
			Collections.sort(list, new Comparator<QueryResultVO>() {
				public int compare(QueryResultVO o1, QueryResultVO o2) {
					return o2.getValue().compareTo(o1.getValue());
				}
			});
			return list;
		}
	}
	
}
